package cn.com.guage.flink.transformation;

import java.util.ArrayList;
import java.util.List;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

//datasource来自集合3                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  
public class StreamFromCollection {

	public static void main(String[] args) throws Exception {
		// 设置流执行环境
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		// 定义一个集合
		List<Person> people = new ArrayList<Person>();
		people.add(new Person("张三", 21));
		people.add(new Person("李四", 16));
		people.add(new Person("王老五", 35));
		people.add(new Person("瓜哥", 33));
		people.add(new Person("猛哥", 15));

		// 加载数据源
		DataStream<Person> flintstones = env.fromCollection(people);
		DataStream<Person> adults = flintstones.filter(new FilterFunction<StreamFromCollection.Person>() {
			
			/**
			 * 
			 */
			private static final long serialVersionUID = 1462502979457828440L;

			public boolean filter(Person value) throws Exception {
				if(value.age>20) {
					return true;
				}
				return false;
			}
		});

		// 对数据执行filter转换
		/*
		 * DataStream<Person> adults = flintstones.filter(new FilterFunction<Person>() {
		 *//**
			* 
			*//*
				 * private static final long serialVersionUID = 1462502979457828440L;
				 * 
				 * public boolean filter(Person person) throws Exception { return person.age >=
				 * 18; } });
				 */

		// 输出转换结果
		adults.print();

		// 触发流程序执行
		env.execute();
	}

	// POJO类
	public static class Person {
		public String name;
		public Integer age;

		public Person() {
		};

		public Person(String name, Integer age) {
			this.name = name;
			this.age = age;
		};

		public String toString() {
			return this.name.toString() + ": age " + this.age.toString();
		};
	}

}
